package com.huitone.smspfm.task;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

import com.huitone.smspfm.logger.ExtLogger;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.ArrayUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.context.app.AppManager;
import com.huitone.smspfm.context.utils.ContextUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.constant.DealResult;
import com.huitone.smspfm.core.filter.SmsRouteFilterTask;
import com.huitone.smspfm.core.waterpipe.client.IGateLink;
import com.huitone.smspfm.core.waterpipe.client.impl.SimpleGateLinkRegister;
import com.huitone.smspfm.model.SmsEntityForMt;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;

/**   
* @Title RebuildPriListManager.java 
* @Package com.huitone.smspfm.module 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年4月24日 下午4:42:29 
* @version V1.0   
*/
public class RebuildPriListTask extends Thread {
	
	private static final ExtLogger logger = ExtLogger.create(RebuildPriListTask.class);
	
	@Override
	public void run() {
		
		logger.info("预处理程序正在检测是否有优先级队列需要重构...");
		
		SimpleGateLinkRegister simpleGateLinkRegister = ContextUtils.getBean2(SimpleGateLinkRegister.class);
		
		Map<String, IGateLink> gateLinks = simpleGateLinkRegister.getGateLinks();
		
		// 需要重构的队列
		Set<String> priListRefactoring = new HashSet<>();
		
		// 组合当前的所有可能优先级集合
		Set<String> currPriList = new HashSet<>();
		if (gateLinks != null) {
			Set<String> linkIds = gateLinks.keySet();
			for (String linkId:linkIds) {
				if (!StringUtils.isEmpty(linkId)) {
					
					Iterator<Entry<String, Integer>> iterator = AppStaticConfigs.configSnodeScodePLvMap.entrySet().iterator();
					
					while (iterator.hasNext()) {
						Entry<String, Integer> entry = iterator.next();
						Integer strLevel = entry.getValue();
//						currPriList.add("SM:" + linkId + ":Pri:" + strLevel + ":List");
						currPriList.add("SM:" + linkId + ":Cache:" + strLevel + ":List");
					}
					
					Integer level = AppStaticConfigs.configMinPLvInteger;
					
					if (level != null) {
//						currPriList.add("SM:" + linkId + ":Pri:" + level + ":List");
						currPriList.add("SM:" + linkId + ":Cache:" + level + ":List");
					}
				}
			}
		}
		
		// 查询redis中已存在的优先级队列
		Cursor<Object> oldPriList = AppStaticConfigs.setOpts.scan(AppStaticConfigs.iSMPriListNamesSet, ScanOptions.NONE);
		while (oldPriList.hasNext()) {
			String priListName = (String) oldPriList.next();
			if (!currPriList.contains(priListName)) {
				priListRefactoring.add(priListName);
			}
		}
		try {
			oldPriList.close();
		} catch (IOException e) {
			logger.error("重建优先级队列任务中,查询已存在优先级队列SMPriListNamesSet后关闭Cursor报错!");
		}
		
		int size = priListRefactoring.size();
		
		if (size > 0) {
			
			logger.info("检测到有" + size + "条优先级队列需要重构,包括:" + ArrayUtils.toString(priListRefactoring));
			
			logger.info("开始重构优先级队列...");
			
			@SuppressWarnings("rawtypes")
			RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetRebuildPri, List.class);
			
			// 重构
			for (String priList:priListRefactoring) {
				
				List<String> keys = new ArrayList<>();
				keys.add(priList);
				keys.add(AppStaticConfigs.pSMCacheHash);
				
				new Thread(new Runnable() {
					
					@Override
					public void run() {
						
						while(AppManager.isRunning()) {
							
							List<List<String>> entities = AppStaticConfigs.scriptOpts.getSMCacheAndHash(script, keys, 1000 + "");
							
							List<SmsEntityForMt> lists = new ArrayList<>();
							for (int i = 0 ; i < entities.size() ; i ++) {
								List<String> list = entities.get(i);
								SmsEntityForMt smsEntity = SmsEntityForMt.parse(list);
								lists.add(smsEntity);
							}
							
							if (!lists.isEmpty()) {
								
								logger.debug("从需要重建的优先级队列" + priList + "中读取到" + lists.size() + "条短信");
								
								CyclicBarrier barrier = new CyclicBarrier(2);
								
								new SmsRouteFilterTask(lists, barrier,false).start();
								
								try {
									barrier.await();
								} catch (InterruptedException | BrokenBarrierException e) {
									e.printStackTrace();
								}
								
								new Thread(new Runnable() {
									
									@Override
									public void run() {
										
										AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

											@Override
											public void excute(RedisConnection connection) {
												
												Set<String> tmpSet = new HashSet<String>();
												
												for (SmsEntityForMt entity:lists) {
													
													String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
													
													Map<byte[], byte[]> map = entity.getMapB();
													
													String priListName = entity.getPriListName();
													
													if (!StringUtils.isEmpty(priListName)) {
														// 路由成功
														map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT0B);
														map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT0_DESCB);
														map.put(AppCanstant.SmsEntity.GateLinkId, AppStaticConfigs.stringRedisSerializer.serialize(entity.getGateLinkId()));
														
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
														
														byte[] priListNameB = AppStaticConfigs.stringRedisSerializer.serialize(priListName);
														
														connection.lPush(priListNameB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
														
														if (!tmpSet.contains(priListName)) {
															tmpSet.add(priListName);
															connection.sAdd(AppStaticConfigs.iSMPriListNamesSetB, priListNameB);
														}
														
														GetPriListLengthTask.priListLength.put(priListName, StringUtils.toLong(GetPriListLengthTask.priListLength.get(priListName)) + 1);
														
														logger.info("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]重构优先级队列路由到" + priListName);
													
													} else {
														// 路由失败
														map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_9B);
														map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_9_DESCB);
														
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
														
														connection.lPush(AppStaticConfigs.iSMRouteFailedListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
														
														connection.lPush(AppStaticConfigs.iSMNotifyListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum + ":2"));
													
														logger.warn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]重构优先级队列时路由失败.");
													}
												}
											}
										});
									}
								}).start();
								
							} else {
								
								break;
							}
						}
					}
				}).start();
			}
			logger.info("优先级队列重构完成!");
		} else {
			logger.info("没有检测到需要重构的优先级队列");
		}
	}
}
